package felix.things.router.opc;

import felix.things.router.config.OpcUaConstant;
import felix.things.router.opc.conf.OpcUaServerConfiguration;
import felix.things.router.opc.conf.mapping.DeviceMapping;
import felix.things.router.opc.external.DefaultMonitorWatcher;
import felix.things.router.opc.external.MonitorWatcher;
import felix.things.router.opc.rpc.RpcProcessor;
import felix.things.router.opc.scan.OpcUaNode;
import felix.things.router.util.CertificateInfo;
import felix.things.router.util.ConfigurationTools;
import felix.things.router.util.OpcUaUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.OpcUaSession;
import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfig;
import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfigBuilder;
import org.eclipse.milo.opcua.sdk.client.api.identity.IdentityProvider;
import org.eclipse.milo.opcua.sdk.client.api.nodes.VariableNode;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.stack.client.DiscoveryClient;
import org.eclipse.milo.opcua.stack.core.AttributeId;
import org.eclipse.milo.opcua.stack.core.Identifiers;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.QualifiedName;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode;
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
import org.eclipse.milo.opcua.stack.core.types.structured.*;

import java.io.IOException;
import java.security.GeneralSecurityException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;

import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
import static org.eclipse.milo.opcua.stack.core.util.ConversionUtil.toList;

@Slf4j
public class OpcUaServerMonitor implements OpcUaDeviceAware {

    private OpcUaServerConfiguration configuration;
    private OpcUaClient client;
    private UaSubscription subscription;
    private Map<NodeId, List<OpcUaDevice>> devicesByTags;
    private Map<String, OpcUaDevice> devicesByName;
    private Map<String, DeviceMapping> mappings;
    private ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

    private ScheduledExecutorService pollingExecutor = Executors.newSingleThreadScheduledExecutor();

    private boolean isConnect = false;

    ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

    private final AtomicLong clientHandles = new AtomicLong(1L);

    private MonitorWatcher watcher;

    private RpcProcessor rpcProcessor;

    public OpcUaServerMonitor(MonitorWatcher watcher, RpcProcessor rpcProcessor, OpcUaServerConfiguration configuration) {
        if (watcher == null) {
            watcher = new DefaultMonitorWatcher();
        }
        this.rpcProcessor = rpcProcessor;
        this.watcher = watcher;
        this.configuration = configuration;
        this.devicesByTags = new ConcurrentHashMap<>();
        this.mappings = configuration.getMapping().stream().collect(Collectors.toMap(DeviceMapping::getDeviceNodePattern, Function.identity()));
        this.devicesByName = new ConcurrentHashMap<>();
    }

    public void connect() {
        cachedThreadPool.execute(() -> {
            while (!isConnect) {
                try {
                    log.info("Initializing OPC-UA server 【{}】 connection to [{}:{}]!", configuration.getApplicationName(), configuration.getHost(), configuration.getPort());

                    SecurityPolicy securityPolicy = SecurityPolicy.valueOf(configuration.getSecurity());
                    IdentityProvider identityProvider = configuration.getIdentity().toProvider();
                    // 连接
                    List<EndpointDescription> endpoints;
                    String endpointUrl = "opc.tcp://" + configuration.getHost() + ":" + configuration.getPort() + configuration.getApplicationUri();
                    try {
                        endpoints = DiscoveryClient.getEndpoints(endpointUrl).get();
                    } catch (Throwable ex) {
                        // try the explicit discovery endpoint as well
                        String discoveryUrl = endpointUrl;

                        if (!discoveryUrl.endsWith("/")) {
                            discoveryUrl += "/";
                        }
                        discoveryUrl += "discovery";

                        log.info("OPC-UA server 【{}】 Trying explicit discovery URL: {}", configuration.getApplicationName(), discoveryUrl);
                        endpoints = DiscoveryClient.getEndpoints(discoveryUrl).get();
                    }
                    EndpointDescription endpoint = endpoints.stream()
                            .filter(e -> e.getSecurityPolicyUri().equals(securityPolicy.getUri()))
                            .findFirst()
                            .orElseThrow(() -> new Exception("no desired endpoints returned"));

                    log.info("OPC-UA server 【{}】 Using endpoint: {} [{}/{}]",
                            configuration.getApplicationName(), endpoint.getEndpointUrl(), securityPolicy, endpoint.getSecurityMode());

                    OpcUaClientConfig config = getOpcUaClientConfig(securityPolicy, identityProvider, endpoint);

                    client = OpcUaClient.create(config);
                    client.connect().get();

                    isConnect = true;
                } catch (Exception e) {
                    log.error("OPC-UA server 【{}】 connection failed! retry interval 【{}】",
                            configuration.getApplicationName(), configuration.getRetryInterval(), e);
                    if (configuration.getRetryInterval() <= 0) {
                        return;
                    }
                    try {
                        Thread.sleep(configuration.getRetryInterval());
                    } catch (InterruptedException e1) {
                        log.error("OPC-UA server 【{}】 Failed to wait for retry interval!", configuration.getApplicationName(), e);
                    }
                }
            }
            try {
                // 订阅tag更新
                if (OpcUaConstant.TYPE_SUBSCRIBE.equals(configuration.getType())) {
                    subscription = client.getSubscriptionManager().createSubscription(1000.0).get();
                }
                // 轮询tag更新
                if (OpcUaConstant.TYPE_POLLING.equals(configuration.getType())) {
                    // 开启轮询线程
                    pullingDevices();
                }
                // 关联 RpcProcessor
                rpcProcessor.putProcessor(configuration.getApplicationName(), watcher, client, this);
                // 深度递归并开启定时扫描
                scanForDevices(configuration.getRootNode());
            } catch (Exception e) {
                log.error("OPC-UA server 【{}】 subscription failed!", configuration.getApplicationName());
                throw new RuntimeException("OPC-UA server [" + configuration.getApplicationName() + "] subscription failed!", e);
            }
        });
    }

    private OpcUaClientConfig getOpcUaClientConfig(SecurityPolicy securityPolicy, IdentityProvider identityProvider, EndpointDescription endpoint) throws GeneralSecurityException, IOException {

        OpcUaClientConfigBuilder builder = OpcUaClientConfig.builder()
                .setApplicationName(LocalizedText.english(configuration.getApplicationName()))
                .setApplicationUri(configuration.getApplicationUri())

                .setEndpoint(endpoint)
                .setIdentityProvider(identityProvider)
                .setRequestTimeout(uint(configuration.getTimeoutInMillis()));


        if (securityPolicy != SecurityPolicy.None) {
            CertificateInfo certificate = ConfigurationTools.loadCertificate(configuration.getKeystore());
            builder.setCertificate(certificate.getCertificate())
                    .setKeyPair(certificate.getKeyPair());
        }
        return builder.build();
    }

    public void disconnect() {
        if (client != null) {
            log.info("Disconnecting from OPC-UA server!");
            try {
                client.disconnect().get(10, TimeUnit.SECONDS);
                log.info("Disconnected from OPC-UA server!");
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.info("Failed to disconnect from OPC-UA server!");
            }
        }
    }

    public void scanForDevices(String rootNode) {
        try {
            long startTs = System.currentTimeMillis();
            NodeId parse = null;
            try {
                parse = NodeId.parse(rootNode);
            } catch (Exception e) {
                log.error("rootNode格式错误！ 使用默认的根节点！");
                parse = Identifiers.ObjectsFolder;
            }
            log.debug("test rootNode:[{}]", parse);
            List<OpcUaDevice> deleted = devicesByName.values().stream()
                    .filter(opcUaDevice -> Math.abs(opcUaDevice.getScanTs() - startTs) >= configuration.getScanPeriodInSeconds() * 1000)
                    .collect(Collectors.toList());
            if (deleted.size() > 0) {
                log.info("Devices {} are no longer available", deleted);
            }
            deleted.forEach(opcUaDevice -> {
                devicesByName.remove(opcUaDevice.getDeviceName());
                try {
                    // 取消订阅
                    subscription.deleteMonitoredItems(opcUaDevice.getSubscribeTags()).get();
                    opcUaDevice.getSubscribeTags().clear();
                    // 设备断开连接
                    watcher.deviceDisConnect(configuration.getApplicationName(), opcUaDevice);
                } catch (InterruptedException | ExecutionException e) {
                    log.error("device tag unsubscribe failed!", e);
                }
            });
            // 当前子节点深度递归
            scanForDevices(new OpcUaNode(parse, ""));
            log.info("Device scan cycle completed in {} ms", (System.currentTimeMillis() - startTs));
        } catch (Exception e) {
            log.warn("Periodic device scan failed!", e);
        }

        log.info("Scheduling next scan in {} seconds!", configuration.getScanPeriodInSeconds());
        if (configuration.getScanPeriodInSeconds() > 0) {
            executor.schedule(() -> {
                scanForDevices(configuration.getRootNode());
            }, configuration.getScanPeriodInSeconds(), TimeUnit.SECONDS);
        }
    }

    public void pullingDevices() {
        try {
            long startTs = System.currentTimeMillis();
            // 按设备轮询
            ArrayList<CompletableFuture<?>> futureList = new ArrayList<>();
            devicesByName.values().forEach(opcUaDevice -> {
                List<NodeId> nodeIds = new ArrayList<>(opcUaDevice.getTagIdsMap().keySet());
                CompletableFuture<List<DataValue>> future = client.readValues(0.0, TimestampsToReturn.Both, nodeIds);
                futureList.add(future.thenAccept(dataValues -> {
                    Map<String, Object> attributes = new HashMap<>();
                    Map<String, Object> timeseries = new HashMap<>();
                    for (int i = 0; i < nodeIds.size(); i++) {
                        DataValue dataValue = dataValues.get(i);
                        NodeId nodeId = nodeIds.get(i);
                        opcUaDevice.updateTag(nodeId, dataValue);
                        attributes.putAll(opcUaDevice.getAffectedAttributes(nodeId, dataValue));
                        timeseries.putAll(opcUaDevice.getAffectedTimeseries(nodeId, dataValue));
                    }
                    // 聚合返回数据
                    if (attributes.size() > 0) {
                        // 更新attributes数据
                        log.trace("OPC-UA server 【{}】 device 【{}】/【{}】 attributes 【{}】",
                                configuration.getApplicationName(),
                                opcUaDevice.getDeviceType(), opcUaDevice.getDeviceName(), attributes);
                        watcher.attributesPoll(configuration.getApplicationName(), opcUaDevice, attributes);
                    }
                    if (timeseries.size() > 0) {
                        // 更新timeseries数据
                        log.trace("OPC-UA server 【{}】 device 【{}】/【{}】 timeseries 【{}】",
                                configuration.getApplicationName(),
                                opcUaDevice.getDeviceType(), opcUaDevice.getDeviceName(), timeseries);
                        watcher.timeseriesPoll(configuration.getApplicationName(), opcUaDevice, timeseries);
                    }
                }));
            });
            CompletableFuture<Void> all = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]));
            if (!all.isDone()) {
                //阻塞  一直等待线程结束
                all.get();
            }
            log.debug("【{}】Device pulling cycle completed in {} ms", devicesByName.size(), (System.currentTimeMillis() - startTs));
        } catch (Exception e) {
            log.warn("Periodic device pulling failed!", e);
        }

        log.debug("Devices pulling next in {} seconds!", configuration.getPollingInSeconds());
        if (configuration.getPollingInSeconds() > 0) {
            pollingExecutor.schedule(() -> {
                // 修复定时运行吞异常
                try {
                    pullingDevices();
                } catch (Exception e) {
                    log.error("device pulling failed!", e);
                }
            }, configuration.getPollingInSeconds(), TimeUnit.SECONDS);
        }
    }

    @Override
    public OpcUaDevice getDevice(String deviceName) {
        return devicesByName.get(deviceName);
    }

    private void scanForDevices(OpcUaNode node) {
        if (node.getParent() != null) {
            log.trace("Scanning node: {},parent node 【{}】", node, "ns=" + node.getParent().getNodeId().getNamespaceIndex() + ";s=" + node.getParent().getNodeId().getIdentifier().toString());
        } else {
            log.trace("Scanning node: {}", node);
        }
        List<DeviceMapping> matchedMappings = mappings.entrySet().stream()
                .filter(
                        mappingEntry -> {
                            String nodeId = node.getNodeId().getIdentifier().toString();
                            if (nodeId.startsWith(mappingEntry.getKey())) {
                                Optional<Integer> max = Arrays.stream(mappingEntry.getValue().getDeviceNameIndex().split(",")).map(Integer::valueOf).max(Integer::compareTo);
                                if (max.isPresent()) {
                                    // P/BXP/TestType/AA/abc
                                    // split:  p bxp testtype aa abc  // length: 5
                                    // max: 3
                                    String[] split = nodeId.split(mappingEntry.getValue().getDeviceSeparator());
                                    // 只需要设备名后面加属性的node
                                    if (split.length - 1 == max.get() + mappingEntry.getValue().getDeviceAttrOffset()) {
                                        return true;
                                    }
                                    log.debug("maxSeparator: {},nodeId: {}", max.get() + mappingEntry.getValue().getDeviceAttrOffset(), nodeId);
                                    return false;
                                }
                            }
                            return false;
                        }
                )
                .map(Map.Entry::getValue).collect(Collectors.toList());
        log.trace("matchedMappings: {}", matchedMappings);
        matchedMappings.forEach(m -> {
            try {
                // 扫描当前设备下的tag
                scanDevice(node, m);
            } catch (Exception e) {
                log.error("Failed to scan device: {}", node.getName(), e);
            }
        });

        try {
            BrowseResult browseResult = client.browse(OpcUaUtils.getBrowseDescription(node.getNodeId())).get();
            List<ReferenceDescription> references = toList(browseResult.getReferences());
            log.trace("references: {}", references);
            for (ReferenceDescription rd : references) {
                NodeId nodeId;
                if (rd.getNodeId().isLocal()) {
                    nodeId = rd.getNodeId().local().get();
                } else {
                    log.trace("Ignoring remote node: {}", rd.getNodeId());
                    continue;
                }
                OpcUaNode childNode = new OpcUaNode(node, nodeId, rd.getBrowseName().getName());

                // recursively browse to children
                scanForDevices(childNode);
            }
        } catch (InterruptedException | ExecutionException e) {
            log.error("Browsing nodeId={} failed: {}", node, e.getMessage(), e);
        }
    }

    private void scanDevice(OpcUaNode node, DeviceMapping m) throws Exception {
        if (node.getParent() != null) {
            log.trace("Scanning device node: {},device parent node 【{}】", node, "ns=" + node.getParent().getNodeId().getNamespaceIndex() + ";s=" + node.getParent().getNodeId().getIdentifier().toString());
        } else {
            log.trace("Scanning device node: {}", node);
        }
        Set<String> tags;
        if (m.isAutoScan()) {
            tags = Collections.emptySet();
            log.trace("Scanning node hierarchy all tags");
        } else {
            tags = m.getAllTags();
            log.trace("Scanning node hierarchy for tags: {}", tags);
        }
        Map<String, NodeId> tagMap = new HashMap<>();
//        Map<String, NodeId> tagMap = OpcUaUtils.lookupTags(client, node.getNodeId(), node.getName(), tags);
        String nodeIds = node.getNodeId().getIdentifier().toString();
        tagMap.put(nodeIds.substring(nodeIds.lastIndexOf(m.getDeviceSeparator()) + 1), node.getNodeId());
        log.trace("Scanned {} tags out of {}", tagMap.size(), tags.size());

        // 构建设备对象
        String[] split = node.getNodeId().getIdentifier().toString().split(m.getDeviceSeparator());
        // 此处构建设备名 X_XX_XXX_XX
        List<Integer> ids = new ArrayList<>();
        if (m.getDeviceNameIndex().contains(",")) {
            ids = Arrays.stream(m.getDeviceNameIndex().split(",")).map(Integer::valueOf).collect(Collectors.toList());
        } else {
            ids.add(Integer.valueOf(m.getDeviceNameIndex()));
        }
        StringBuilder deviceName = new StringBuilder(split[ids.get(0)]);
        for (int i = 1; i < ids.size() && i < split.length; i++) {
            deviceName.append("_").append(split[ids.get(i)]);
        }

        OpcUaDevice device;
        if (devicesByName.containsKey(deviceName.toString())) {
            device = devicesByName.get(deviceName.toString());
        } else {
            device = new OpcUaDevice(deviceName.toString(), m);
            try {
                // 修复数组下标越界
                String deviceType = split[m.getDeviceTypeIndex()];
                device.setDeviceType(deviceType);
            } catch (Exception ignored) {
            }
            devicesByName.put(device.getDeviceName(), device);

            // 设备连接创建
            watcher.deviceConnect(configuration.getApplicationName(), device);
        }

        // 取消扫描更新时间，改为数据上传更新时间
//        device.updateScanTs();

        Map<String, NodeId> newTags = device.registerTags(tagMap);
        if (newTags.size() > 0) {
            for (NodeId tagId : newTags.values()) {
                devicesByTags.putIfAbsent(tagId, new ArrayList<>());
                devicesByTags.get(tagId).add(device);
            }
            if (OpcUaConstant.TYPE_SUBSCRIBE.equals(configuration.getType())) {
                log.debug("Going to subscribe to tags: {}", newTags);
                subscribeToTags(device, newTags);
            }
        }
    }

    /**
     * @param device
     * @param newTags
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private void subscribeToTags(OpcUaDevice device, Map<String, NodeId> newTags) throws InterruptedException, ExecutionException {
        List<MonitoredItemCreateRequest> requests = new ArrayList<>();
        for (Map.Entry<String, NodeId> kv : newTags.entrySet()) {
            // subscribe to the Value attribute of the server's CurrentTime node
            ReadValueId readValueId = new ReadValueId(
                    kv.getValue(),
                    AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE
            );

            // IMPORTANT: client handle must be unique per item within the context of a subscription.
            // You are not required to use the UaSubscription's client handle sequence; it is provided as a convenience.
            // Your application is free to assign client handles by whatever means necessary.
            UInteger clientHandle = subscription.nextClientHandle();

            MonitoringParameters parameters = new MonitoringParameters(
                    clientHandle,
                    1000.0,     // sampling interval
                    null,       // filter, null means use default
                    uint(10),   // queue size
                    true        // discard oldest
            );

            requests.add(new MonitoredItemCreateRequest(
                    readValueId,
                    MonitoringMode.Reporting,
                    parameters
            ));

        }

        BiConsumer<UaMonitoredItem, Integer> onItemCreated =
                (item, id) -> item.setValueConsumer(this::onSubscriptionValue);

        log.debug("SubscribeRequestsSize={}", requests.size());
        List<UaMonitoredItem> items = null;
        try {
            items = subscription.createMonitoredItems(
                    TimestampsToReturn.Both,
                    requests,
                    onItemCreated
            ).get();
        } catch (InterruptedException e) {
            log.warn("设备 {} 订阅服务超时", device.getDeviceName());
            device.unRegisterTags(newTags);
            return;
        } catch (ExecutionException e) {
            log.warn("设备 {} 订阅服务端数据失败：{}", device.getDeviceName(), newTags.values());
            device.unRegisterTags(newTags);
            return;
        }
        for (UaMonitoredItem item : items) {
            // 保存订阅tag，用于离线时删除
            device.getSubscribeTags().add(item);
            if (item.getStatusCode().isGood()) {
                log.trace("Monitoring Item created for nodeId={}", item.getReadValueId().getNodeId());
            } else {
                log.warn("Failed to create item for nodeId={} (status={})",
                        item.getReadValueId().getNodeId(), item.getStatusCode());
            }
        }
    }

    private void onSubscriptionValue(UaMonitoredItem item, DataValue dataValue) {
        NodeId tagId = item.getReadValueId().getNodeId();
        log.debug("Subscription value received: item={}, value={}",
                tagId, dataValue.getValue());
        devicesByTags.getOrDefault(tagId, Collections.emptyList()).forEach(
                device -> {
                    // 有数据订阅更新时，刷新最后时间，用于判断是否离线
                    device.updateScanTs();
                    device.updateTag(tagId, dataValue);
                    Map<String, Object> attributes = device.getAffectedAttributes(tagId, dataValue);
                    if (attributes.size() > 0) {
                        // 更新attributes数据
                        log.trace("OPC-UA server 【{}】 device 【{}】/【{}】 attributes 【{}】",
                                configuration.getApplicationName(),
                                device.getDeviceType(), device.getDeviceName(), attributes);
                        watcher.attributesUpdated(configuration.getApplicationName(), device, attributes);
                    }
                    Map<String, Object> timeseries = device.getAffectedTimeseries(tagId, dataValue);
                    if (timeseries.size() > 0) {
                        // 更新timeseries数据
                        log.trace("OPC-UA server 【{}】 device 【{}】/【{}】 timeseries 【{}】/【{}】",
                                configuration.getApplicationName(),
                                device.getDeviceType(), device.getDeviceName(), dataValue.getServerTime(), timeseries);
                        watcher.timeseriesUpdated(configuration.getApplicationName(), device, timeseries);
                    }
                }
        );
    }

    private Map<String, String> readTags(Map<String, NodeId> tags) throws ExecutionException, InterruptedException {
        Map<String, CompletableFuture<DataValue>> dataFutures = new HashMap<>();
        for (Map.Entry<String, NodeId> kv : tags.entrySet()) {
            VariableNode node = client.getAddressSpace().createVariableNode(kv.getValue());
            dataFutures.put(kv.getKey(), node.readValue());
        }

        Map<String, String> result = new HashMap<>();
        for (Map.Entry<String, CompletableFuture<DataValue>> kv : dataFutures.entrySet()) {
            String tag = kv.getKey();
            DataValue value = kv.getValue().get();
            result.put(tag, value.getValue().getValue().toString());
        }
        return result;
    }

    public void checkConnection() {
        cachedThreadPool.execute(() -> {
            while (true) {
                if (isConnect) {
                    try {
                        OpcUaSession opcUaSession = client.getSession().get();
                        log.debug("opcua session is {}", opcUaSession.getSessionId());
                    } catch (Exception e) {
                        log.error("connect is not alive ! 【{}】", e.getMessage());
                        isConnect = false;
                        unbindDevice();
                        connect();
                    }
                }
                try {
                    Thread.sleep(configuration.getRetryInterval());
                } catch (InterruptedException e) {
                    log.error("OPC-UA server 【{}】 Failed to wait for retry interval!", configuration.getApplicationName(), e);
                }
            }
        });
    }

    /**
     * 删除原有绑定数据
     */
    private void unbindDevice() {
        devicesByName.clear();
        devicesByTags.clear();
    }
}
